package job;

import function.CarViewBeanMapFuction;
import function.CarViewWindowFunction;
import net.bwie.realtime.jtp.common.utils.DorisUtil;
import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;

import java.time.Duration;

public class CarAppLogToDoris {
    public static void main(String[] args) throws Exception {
        // 1.env执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // 每三秒打一次快照
        env.enableCheckpointing(3000L);
        // 2.消费kafka
        DataStream<String> kafkaStream = KafkaUtil.consumerKafka(env, "dwd-car-logs");
//        kafkaStream.print("Kafka");
        // 3.数据转换
        DataStream<String> resultStream = handle(kafkaStream);
        resultStream.print();
        // 4.数据输出
        DorisUtil.saveToDoris(
                resultStream
                ,"ner_car_report"
                ,"dwd_car_status_info"
        );
        // 5.触发执行
        env.execute("CarAppLogToDoris");
    }

    private static DataStream<String> handle(DataStream<String> stream) {
        // 第一步:将流中的每一条数据封装实体类
        SingleOutputStreamOperator<CarBean> beanStream = stream.map(new CarViewBeanMapFuction());
        // 第二步:设置水位线
        SingleOutputStreamOperator<CarBean> timeStream = beanStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<CarBean>forBoundedOutOfOrderness(Duration.ofSeconds(0))
                        .withTimestampAssigner(new SerializableTimestampAssigner<CarBean>() {
                            @Override
                            public long extractTimestamp(CarBean carBean, long l) {
                                return carBean.getTimestamp();
                            }
                        })
        );
        // 第三步:存储所有数据不分组
        KeyedStream<CarBean, String> keyByStream = timeStream.keyBy(bean -> "");
        // 第四步:设置滚动窗口 窗口大小1分钟
        WindowedStream<CarBean, String, TimeWindow> windowStream = keyByStream.window(
                TumblingEventTimeWindows.of(Time.seconds(1))
        );
        // 第五步:我这的作用是转换类型
        SingleOutputStreamOperator<String> applyStream = windowStream.apply(new CarViewWindowFunction());
        return applyStream;
    }
}
